package com.bw.flinkstreaming.state2.job2;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Collections;

/**
 *
 * 需求：相同key的单词出现三次就计算平均值
 *   hadoop
 *   hive
 *   hive
 *   hadoop
 *   spark
 *   hadoop  -> 触发计算
 *   hive    -> 触发计算
 *   hadoop
 *   hadoop
 *   hadoop  -> 触发计算
 *
 *  自定义状态实现该需求，使用ListState状态
 * */
public class ListStateCountAvg extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Double>> {

    //定义状态
    private ListState<Tuple2<Long,Long>> listCountAvg;

    /**
     * 状态初始化
     * */
    @Override
    public void open(Configuration parameters) throws Exception {
        //注册状态
        ListStateDescriptor descriptor = new ListStateDescriptor(
                "listCountAvg",
                Types.TUPLE(Types.LONG,Types.LONG));
        listCountAvg = getRuntimeContext().getListState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Double>> out) throws Exception {
        //获取当前key对应的历史状态
        Iterable<Tuple2<Long, Long>> currentState = listCountAvg.get();
        if (currentState == null) {
            listCountAvg.addAll(Collections.emptyList());
        }

        //更新状态
        listCountAvg.add(value);

        //判断是否满足三次
        ArrayList<Tuple2<Long, Long>> listsState = Lists.newArrayList(listCountAvg.get());
        if (listsState.size() == 3) {
            long count = 0;
            double sum = 0;
            for(Tuple2<Long,Long> ele:listsState) {
                count++;
                sum += ele.f1;
            }
            double avg = (Double)(sum / count);
            out.collect(Tuple2.of(value.f0,avg));
            //清空
            listCountAvg.clear();

        }
    }
}
